package com.demo.java.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.Data;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

// 多生产者, 多消费者
public class DisruptorMutiProducerMutiConsumerTest {
    @Data
    private static class Element {
        private String value;
        public static Element create() {
            return new Element();
        }
    }
    private static class Consumer implements WorkHandler<Element> {
        public void onEvent(Element element) throws Exception {
            System.out.println("consumer handler: " + element.getValue());
        }
    }
    private static class Producer {
        private RingBuffer<Element> ringBuffer;
        public Producer(RingBuffer<Element> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }
        public void sendData(String value) {
            long sequence = this.ringBuffer.next();
            this.ringBuffer.get(sequence).setValue(value);
            this.ringBuffer.publish(sequence);
        }
    }
    public static void main(String[] args) throws InterruptedException {
        int consumerSize = 10;
        int bufferSize = 1024 * 1024;
        WaitStrategy yieldStrategy = new YieldingWaitStrategy();
        CountDownLatch latch = new CountDownLatch(1);

        EventFactory<Element> eventFactory = new EventFactory<Element>() {
            public Element newInstance() {
                return Element.create();
            }
        };
        ExceptionHandler<Element> exceptionHandler = new ExceptionHandler<Element>() {
            public void handleEventException(Throwable throwable, long l, Element element) {
                throwable.printStackTrace();
            }
            public void handleOnStartException(Throwable throwable) {
                throwable.printStackTrace();
            }
            public void handleOnShutdownException(Throwable throwable) {
                throwable.printStackTrace();
            }
        };
        RingBuffer<Element> ringBuffer = RingBuffer.create(ProducerType.MULTI, eventFactory, bufferSize, yieldStrategy);
        Consumer[] consumers = new Consumer[consumerSize];
        for (int i = 0; i < consumerSize; i++) {
            consumers[i] = new Consumer();
        }
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        WorkerPool<Element> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, exceptionHandler, consumers);
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        workerPool.start(Executors.newFixedThreadPool(20));

        // 创建10个生产者
        for (int i = 0; i < 2; i++) {
            new Thread("producer-thread-" + i) {
                public void run() {
                    Producer producer = new Producer(ringBuffer);
                    for (int j = 0; true; j++) {
                        producer.sendData(Thread.currentThread().getName() + ", seq=" + j);
                    }
                }
            }.start();
        }

        latch.await();
    }
}
